package com.xsomnus.rpc.consumer.net;

import com.xsomnus.rpc.consumer.netty.NettyClient;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author @xsomnus666_xiawenye★
 * @since 2019/4/22 0022 16:57
 * 当你发现自己的才华撑不起野心时，就请安静下来学习吧
 */
@Slf4j
@Component
public class DefaultConnectionManager implements ConnectionManager {

    private AtomicInteger roundRobin = new AtomicInteger(0);
    private Map<SocketAddress, Channel> channelNodes = new ConcurrentHashMap<>();
    private CopyOnWriteArrayList<Channel> channels = new CopyOnWriteArrayList<>();
    private final NettyClient nettyClient;

    public DefaultConnectionManager(NettyClient nettyClient) {
        this.nettyClient = nettyClient;
    }

    @Override
    public Channel chooseChannel() {
        if (channels.size() > 0) {
            int size = channels.size();
            int idx = (roundRobin.getAndAdd(1) + size) % size;
            return channels.get(idx);
        }
        return null;
    }

    @Override
    public synchronized void refreshConnectedServer(List<String> addressList) {
        if (addressList == null || addressList.size() == 0) {
            log.error("没有可用的服务器节点, 全部服务节点已关闭!");
            for (final Channel channel : channels) {
                SocketAddress remotePeer = channel.remoteAddress();
                Channel handlerNode = channelNodes.get(remotePeer);
                handlerNode.close();
            }
            channels.clear();
            channelNodes.clear();
            return;
        }

        HashSet<SocketAddress> allServerNodes = new HashSet<>();
        for (int i = 0; i < addressList.size(); i++) {
            String[] arr = addressList.get(i).split(":");
            if (arr.length == 2) {
                String host = arr[0];
                int port = Integer.parseInt(arr[1]);
                final SocketAddress remotePeer = new InetSocketAddress(host, port);
                allServerNodes.add(remotePeer);
            }
        }

        for (final SocketAddress serverNodeAddress :
                allServerNodes) {
            Channel channel = channelNodes.get(serverNodeAddress);
            if (channel != null && channel.isOpen()) {
                log.debug("当前节点已经存在, 无需重新连接: {}", serverNodeAddress);
            } else {
                connectServerNode(serverNodeAddress);
            }
        }


        for (int i = 0; i < channels.size(); ++i) {
            Channel channel = channels.get(i);
            SocketAddress remotePeer = channel.remoteAddress();
            if (!allServerNodes.contains(remotePeer)) {
                log.debug("删除失效的服务节点: {}", remotePeer);
                Channel node = channelNodes.get(remotePeer);
                if (node != null) {
                    node.close();
                }
                channels.remove(channel);
                channelNodes.remove(remotePeer);
            }
        }

    }

    @Override
    public void connectServerNode(SocketAddress socketAddress) {
        try {
            Channel channel = nettyClient.doConnect(socketAddress);
            addChannel(channel, socketAddress);
        } catch (InterruptedException e) {
            e.printStackTrace();
            log.debug("未能成功的连接到RPC服务器: {}", socketAddress);
        }
    }

    @Override
    public void addChannel(Channel channel, SocketAddress socketAddress) {
        log.debug("添加channel到连接管理器中: {}", socketAddress);
        channels.add(channel);
        channelNodes.put(socketAddress, channel);
    }

    @Override
    public void removeChannel(Channel channel) {
        log.debug("从连接管理器中移除失效的channel: {}", channel.remoteAddress());
        SocketAddress remotePeer = channel.remoteAddress();
        channelNodes.remove(remotePeer);
        channels.remove(channel);
    }
}
